/*
 *  Copyright 2020-2025 the original author or authors.
 *  You cannot use this file unless authorized by the author.
 */

package org.ipig.computing.spark.streaming;

import lombok.extern.slf4j.Slf4j;
import org.ipig.commons.helper.GsonHelper;
import org.ipig.commons.helper.StringHelper;
import org.ipig.computing.commons.streaming.StreamExecutorService;
import org.ipig.computing.spark.conf.SparkExecutorConf;

/**
 * 通用的 Stream Job Launcher
 *
 * @author <a href="mailto:comchnts@163.com">chinats</a>
 * @since 1.0
 */
@Slf4j
public class SparkStreamExecutorLauncher {
    public static void main(String[] args) {
        if (args == null || args.length < 1) {
            log.warn("缺少参数！");
            return;
        }
        String json = StringHelper.decodeByUTF8(args[0]);
        StreamExecutorService executorService=null;
        try {
            log.warn("参数：{}", json);
            SparkExecutorConf conf = GsonHelper.getInstance().fromJson(json, SparkExecutorConf.class);
            if (conf != null) {
                Class executorClass = Class.forName(conf.getExecutorClass());
                Object executorInstance = executorClass.newInstance();
                if (StreamExecutorService.class.isInstance(executorInstance)) {
                    executorService = (StreamExecutorService) executorInstance;
                    executorService.setConf(conf);
                    executorService.execute();
                }
            } else {
                log.warn("Spark Stream 配置转化成null");
            }
        } catch (Exception e) {
            log.error(e.getMessage());
            e.printStackTrace();
        }finally {
            if (executorService != null) {
                log.warn("执行器{}执行结束。",executorService.getExecutorClass());
            }
        }
    }
}

